package com.zhang.spark_1.spark_core.wordCount

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/4 21:04
 */
object Spark04_WordCount {
  def main(args: Array[String]): Unit = {
    //获取spark连接

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount4")
    val sc: SparkContext = new SparkContext(conf)
    wordCount9(sc)

    sc.stop()
  }

  //groupBy
  def wordCount1(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))

    val groupWord: RDD[(String, Iterable[String])] = words.groupBy(word => word)
    val wordCount: RDD[(String, Int)] = groupWord.mapValues(iter => iter.size)
    wordCount.collect().foreach(println)
  }

  //groupByKey
  def wordCount2(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordOne: RDD[(String, Int)] = words.map((_, 1))
    val groupWord: RDD[(String, Iterable[Int])] = wordOne.groupByKey()
    val wordCount: RDD[(String, Int)] = groupWord.mapValues(iter => iter.size)
    wordCount.collect().foreach(println)
  }

  //reduceByKey
  def wordCount3(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordOne: RDD[(String, Int)] = words.map((_, 1))
    val wordCount: RDD[(String, Int)] = wordOne.reduceByKey(_ + _)
    wordCount.collect().foreach(println)
  }

  //aggregateByKey
  def wordCount4(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordOne: RDD[(String, Int)] = words.map((_, 1))
    val wordCount: RDD[(String, Int)] = wordOne.aggregateByKey(0)(_ + _, _ + _)
    wordCount.collect().foreach(println)
  }


  //foldByKey
  def wordCount5(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordOne: RDD[(String, Int)] = words.map((_, 1))
    val wordCount: RDD[(String, Int)] = wordOne.foldByKey(0)(_ + _)
    wordCount.collect().foreach(println)
  }

  //combineByKey
  def wordCount6(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordOne: RDD[(String, Int)] = words.map((_, 1))
    val wordCount: RDD[(String, Int)] = wordOne.combineByKey(
      v => v,
      (x: Int, y) => y + x,
      (x: Int, y: Int) => x + y
    )
    wordCount.collect().foreach(println)
  }
  //countByKey
  def wordCount7(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordOne: RDD[(String, Int)] = words.map((_, 1))
    val wordCount: collection.Map[String, Long] = wordOne.countByKey()
    wordCount.foreach(println)
  }

  //countByKey
  def wordCount8(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordCount: collection.Map[String, Long] = words.countByValue()
    wordCount.foreach(println)
  }

  //reduce
  def wordCount9(sc: SparkContext) = {
    val rdd: RDD[String] = sc.makeRDD(List("hello scala", "hello spark", "hello flink"))

    val words: RDD[String] = rdd.flatMap(_.split(" "))
    val wordMap: RDD[mutable.Map[String, Int]] = words.map(
      word => {
        mutable.Map[String, Int]((word, 1))
      }
    )

    val wordCount: mutable.Map[String, Int] = wordMap.reduce(
      (map1, map2) => {
        map2.foreach {
          case (word, v) => {
            val count = map1.getOrElse(word, 0) + v
            map1.update(word, count)
          }
        }
        map1
      }
    )
    wordCount.foreach(println)
  }



}
